package com.lunatic.rocketmq.service;


import com.lunatic.rocketmq.model.DemoMessage;
import lombok.extern.log4j.Log4j2;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;

@Log4j2
@Service
public class ProducerService {

    // 普通消息主题
    private static final String NOR_DEMO_TOPIC = "normal-demo-topic";
    // 事务消息主题
    private static final String TX_DEMO_TOPIC = "tx-demo-topic";

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 普通发送
     */
    public void send(DemoMessage demoMessage) {
        Message<DemoMessage> msg = MessageBuilder
                .withPayload(demoMessage)
                .setHeader(RocketMQHeaders.KEYS, demoMessage.getId())
                .setHeader(RocketMQHeaders.TAGS, "TagN")  // 设置Tag过滤
                .build();
        // 消息头设置Tag的方式无效。采用主题连接Tag的方式。
        rocketMQTemplate.send(NOR_DEMO_TOPIC, msg);
        log.info("普通消息发送成功");
    }


    /**
     * 同步发送消息
     *
     * @param demoMessage
     * @return
     */
    public String sendSyncMessage(DemoMessage demoMessage) {
        // 构造消息，指定Topic和Tag
        Message<DemoMessage> msg = MessageBuilder
                .withPayload(demoMessage)
                .setHeader(RocketMQHeaders.KEYS, demoMessage.getId())
                .setHeader(RocketMQHeaders.TAGS, "TagN")  // 设置Tag过滤
                .build();
        SendResult result = rocketMQTemplate.syncSend(NOR_DEMO_TOPIC, msg);
        log.info("同步发送成功，消息ID：{}", result.getMsgId());
        return result.getMsgId();
    }

    /**
     * 异步发送消息
     *
     * @param demoMessage
     * @return
     */
    public void sendAsyncMessage(DemoMessage demoMessage) {
        Message<DemoMessage> msg = MessageBuilder
                .withPayload(demoMessage)
                .setHeader(RocketMQHeaders.KEYS, demoMessage.getId())
                .setHeader(RocketMQHeaders.TAGS, "TagN")  // 设置Tag过滤
                .build();
        // 异步发送并注册回调
        rocketMQTemplate.asyncSend(NOR_DEMO_TOPIC, msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("异步发送成功：{}", sendResult.getMsgId());
            }

            @Override
            public void onException(Throwable e) {
                log.info("异步发送失败：{}", e.getMessage());
            }
        });
    }


    /**
     * 发送事务消息
     *
     * @param demoMessage
     * @return
     */
    public LocalTransactionState sendTransactionMessage(DemoMessage demoMessage) {
        Message<DemoMessage> msg = MessageBuilder
                .withPayload(demoMessage)
                .setHeader(RocketMQHeaders.KEYS, demoMessage.getId())
                .setHeader(RocketMQHeaders.TAGS, "TagN")  // 设置Tag过滤
                .build();
        // 发送事务消息
        TransactionSendResult result = rocketMQTemplate
                .sendMessageInTransaction(TX_DEMO_TOPIC, msg, null);
        log.info("事务消息状态：{}", result.getLocalTransactionState());
        return result.getLocalTransactionState();
    }
}
